/*
 * 文件名：LTESdtpAcceptedHandler.java
 * 版权：亚信科技(南京)版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2015-12-14
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.protocal.sdtp.netty.handler;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.handler.timeout.ReadTimeoutException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.net.NettyUtil;
import com.ailk.bigdata.etl.realstream.server.constant.ProtocalSdtpConstant;
import com.ailk.bigdata.etl.realstream.server.model.Header;
import com.ailk.bigdata.etl.realstream.server.model.LTENetPackage;
import com.ailk.bigdata.etl.realstream.server.model.MessageType;
import com.ailk.bigdata.etl.realstream.server.model.SdtpDataPack;
import com.ailk.bigdata.etl.realstream.server.model.ValueEvent;
import com.ailk.bigdata.etl.realstream.server.monitor.EventCounter;
import com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.ProtocalConnectMgrService;
import com.ailk.bigdata.etl.realstream.server.protocal.sdtp.server.impl.ProtocalConnectMgrServiceImpl;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;
import com.google.common.primitives.UnsignedInteger;
import com.lmax.disruptor.RingBuffer;

/**
 * @description TODO
 * @author [napo]
 * @version [版本号,2015-12-14]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
@Sharable
public class LTESdtpAcceptedHandler extends SimpleChannelInboundHandler<LTENetPackage>
{

	private static Logger logger = LoggerFactory.getLogger(LTESdtpAcceptedHandler.class);
	private ProtocalConnectMgrService connectMgrService;
	private EventCounter eventCounter;

	public LTESdtpAcceptedHandler(EventCounter eventCounter)
	{
		super();
		this.eventCounter = eventCounter;
		connectMgrService = new ProtocalConnectMgrServiceImpl(eventCounter);
	}

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelActive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception
	{

		String clientAddr = NettyUtil.getClientAdd(ctx);
		logger.info("get connection from client [" + clientAddr + "]");
		ctx.fireChannelActive();
	}

	/**
	 * @see io.netty.channel.ChannelInboundHandlerAdapter#channelInactive(io.netty.channel.ChannelHandlerContext)
	 */
	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception
	{

		String clientAddr = NettyUtil.getClientAdd(ctx);
		RealStreamTool.AUTHENTICATION_INFO_CACHE.remove(clientAddr);
		RealStreamTool.AUTHORISED_USER_INFO.remove(clientAddr);
		ChannelHandlerContext channelHandler = RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(clientAddr);
		if (channelHandler != null)
		{
			channelHandler.channel().close();
		}
		logger.info("监测到remote[{}]关闭,处理缓存信息", clientAddr);

	}

	/**
	 * @see io.netty.channel.SimpleChannelInboundHandler#channelRead0(io.netty.channel.ChannelHandlerContext,
	 *      java.lang.Object)
	 */
	@Override
	protected void channelRead0(ChannelHandlerContext ctx, LTENetPackage dataPack) throws Exception
	{
		eventCounter.incrementEventReceivedCount();
		String clientAddr = NettyUtil.getClientAdd(ctx);
		int messageType = dataPack.getHeader().getMessageType();
		if (messageType == MessageType.notifyXDRData_Req.getValue())
		{
			if (!RealStreamTool.AUTHORISED_USER_INFO.containsKey(clientAddr))
			{

				ctx.writeAndFlush(buildUserNotAuthPack(dataPack.getHeader().getSequenceId()));
				dataPack = null;

			}
			else
			{
				eventCounter.incrementDatasAcceptedCount();
				RealStreamTool.AUTHORISED_USER_INFO.get(clientAddr).incrementDatasAcceptedCount();

				if (RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.containsKey(dataPack.getDataType()))
				{

					eventCounter.incrementEventReceivedByType(Integer.toString(dataPack.getDataType()));

					RingBuffer<ValueEvent> ringBuffer = RealStreamTool.UNITCODE_ACCEPTED_DISRUPTOR.get(dataPack.getDataType());
					long next = ringBuffer.next();
					try
					{
						ValueEvent event = ringBuffer.get(next);
						event.setDataPackage(dataPack);
					}
					finally
					{
						ringBuffer.publish(next);
					}
				}

			}

		}
		else if (messageType == MessageType.XDRRawDataSend_Req.getValue())
		{
			logger.info("接受到的消息类型[{}]", messageType);
		}
		else switch (MessageType.getFixedType(dataPack.getHeader().getMessageType()))
		{
			case verNego_Req:
				logger.info("客户端[{}]发起版本协商请求.........", NettyUtil.getClientAdd(ctx));
				connectMgrService.versionNego(ctx, dataPack);
				break;
			case linkAuth_Req:
				logger.info("客户端[{}]发起版本鉴权请求.........", NettyUtil.getClientAdd(ctx));
				connectMgrService.linkAuth(ctx, dataPack);
				break;
			case linkRel_Req:
				logger.info("客户端[{}]发起链路释放请求.........", NettyUtil.getClientAdd(ctx));
				connectMgrService.linkRel(ctx, dataPack);
				break;
			case linkDataCheck_Req:
				logger.info("客户端[{}]发起数据校验请求.........", NettyUtil.getClientAdd(ctx));
				connectMgrService.linkDataCheck(ctx, dataPack);
				break;
			default:
				ctx.close();
				logger.error("客户端[{}]发起数据类型错误.........", NettyUtil.getClientAdd(ctx));

		}

	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception
	{
		String clientAddr = NettyUtil.getClientAdd(ctx);

		if (cause instanceof ReadTimeoutException)
		{
			logger.error(ctx.channel().remoteAddress().toString() + " timeout exception happened , caused by " + cause.getMessage());
		}
		else
		{
			logger.error(ctx.channel().remoteAddress().toString() + "receive message exception happened ,caused by " + cause, cause);
		}

		RealStreamTool.AUTHORISED_USER_INFO.remove(clientAddr);

		ChannelHandlerContext channelHandler = RealStreamTool.ACCEPTING_CHANNEL_CACHE.remove(clientAddr);
		if (channelHandler != null)
		{
			channelHandler.channel().close();
		}

		logger.info("监测到remote[" + ctx.channel().remoteAddress() + "]连接异常关闭,处理缓存信息");

		ctx.close();
	}

	private SdtpDataPack buildUserNotAuthPack(UnsignedInteger sequence)
	{
		Header header = new Header();
		header.setMessageType(MessageType.linkAuth_Resp.getValue());
		header.setSequenceId(sequence);
		header.setTotalLength(74);
		header.setTotalContents((short) 1);

		SdtpDataPack returnPack = new SdtpDataPack();
		returnPack.setHeader(header);
		ByteBuf buf = Unpooled.buffer();
		buf.writeByte(ProtocalSdtpConstant.SDTP_PROTOCAL_LOGIN_AUTH_SHA_ENCRYPT_ERR);
		returnPack.setDataContent(buf);

		return returnPack;
	}

}
